import com.leilei.Location;
import com.leilei.count.CountKeyWindowFunction;
import com.leilei.count.MyLocationSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.nio.charset.StandardCharsets;


/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 21:55
 * @desc flink 窗口 根据数量开窗口
 */
public class FlinkWindow_CountWindow_RabbitConnectors {
    public static void main(String[] args) throws Exception {

        final RMQConnectionConfig rabbitConfig = new RMQConnectionConfig.Builder()
                .setHost("xxx")
                .setUserName("admin")
                .setPassword("xxx")
                .setPort(5672)
                .setVirtualHost("/")
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        DataStreamSource<Location> source = env.addSource(new MyLocationSource());
        WindowedStream<Location, Integer, GlobalWindow> window = source.keyBy(Location::getVehicleId)
                .countWindow(1);
        SingleOutputStreamOperator<String> result = window.apply(new CountKeyWindowFunction());
        result.addSink(new RMQSink<String>(
                rabbitConfig,
                "over-speeding-alarm",
                new SimpleStringSchema(StandardCharsets.UTF_8)));
        env.execute("rabbit-connectors-window");
    }
}